package com.superhelper.processor.order.consumer;

import java.math.BigDecimal;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.superhelper.common.Consts;
import com.superhelper.common.domain.AuserShop;
import com.superhelper.common.domain.POIStatus;
import com.superhelper.common.domain.POIStatusOperator;
import com.superhelper.common.domain.PushMsgType;
import com.superhelper.common.domain.TakeOutOrder;
import com.superhelper.common.domain.TakeOutOrderBackup;
import com.superhelper.common.domain.TakeOutOrderGroup;
import com.superhelper.common.domain.TakeOutOrderGroupItem;
import com.superhelper.common.domain.TakeOutOrderStatus;
import com.superhelper.common.domain.TakeOutOrderType;
import com.superhelper.common.domain.User;
import com.superhelper.common.domain.ele.ElePushEntity;
import com.superhelper.common.utils.GeoUtil;
import com.superhelper.common.utils.JsonHelper;
import com.superhelper.common.utils.JsonObjectUtil;

@Component
public class EleOrderConsumer extends AbstractOrderConsumer implements ApplicationListener<ContextRefreshedEvent> {

    @Value("${aliyun.access.key}")
    private String aliyunAccessKey;
    @Value("${aliyun.access.secret}")
    private String aliyunSecretKey;

    @Value("${aliyun.ele.order.consumer.id}")
    private String consumerId;
    @Value("${order.topic}")
    private String orderTopic;
    @Value("${online}")
    private boolean online;

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        if (event.getApplicationContext().getParent() != null) {
            return;
        }
        if (consumer != null) {
            return;
        }
        init();
        String tag = orderTopic + online + "-ele";
        consumer.subscribe(orderTopic, tag, new MessageListener() {
            public Action consume(Message message, ConsumeContext context) {
                try {
                    String content = new String(message.getBody(), Consts.UTF8_CHARSET);
                    log.info("content is:{}", content);
                    ElePushEntity entity = JsonHelper.fromJson(content, ElePushEntity.class);
                    if (entity == null) {
                        log.error("EleOrderConsumer entity is null");
                        return Action.CommitMessage;
                    }
                    PushMsgType type = PushMsgType.getByEleType(entity.getType());
                    // 下单消息
                    if (type == PushMsgType.ORDER_CREATED) {
                        createOrder(content, entity);
                    } else if (type == PushMsgType.ORDER_ACCEPTED) {
                        dealOrder(TakeOutOrderStatus.VALID.getValue(), entity.getMessage(), type,
                                TakeOutOrderType.EleMe.getValue());
                    } else if (type == PushMsgType.ORDER_COMPLETED) {
                        dealOrder(TakeOutOrderStatus.SETTLED.getValue(), entity.getMessage(), type,
                                TakeOutOrderType.EleMe.getValue());
                    } else if (type == PushMsgType.ORDER_CANCELED) {
                        dealOrder(TakeOutOrderStatus.INVALID.getValue(), entity.getMessage(), type,
                                TakeOutOrderType.EleMe.getValue());
                    } else if (type == PushMsgType.ORDER_REFUND) {
                        dealOrder(TakeOutOrderStatus.REFUNDING.getValue(), entity.getMessage(), type,
                                TakeOutOrderType.EleMe.getValue());
                    } else if (type == PushMsgType.ORDER_APPLY_CANCEL) {
                        dealOrder(TakeOutOrderStatus.APPLY_CANCEL.getValue(), entity.getMessage(), type,
                                TakeOutOrderType.EleMe.getValue());
                    } else if (type == PushMsgType.ORDER_REVOKE_CANCEL) {
                        dealOrder(TakeOutOrderStatus.VALID.getValue(), entity.getMessage(), type,
                                TakeOutOrderType.EleMe.getValue());
                    } else if (type == PushMsgType.ORDER_AGREE_CANCEL) {
                        dealOrder(TakeOutOrderStatus.INVALID.getValue(), entity.getMessage(), type,
                                TakeOutOrderType.EleMe.getValue());
                    } else if (type == PushMsgType.ORDER_REJECT_CANCEL) {
                        dealOrder(TakeOutOrderStatus.REJECT_CANCEL.getValue(), entity.getMessage(), type,
                                TakeOutOrderType.EleMe.getValue());
                    } else if (type == PushMsgType.DELIVER_STATUS) {
                        if (entity.getType() >= 51 && entity.getType() <= 55) {
                            dealOrder(TakeOutOrderStatus.DELIVERING.getValue(), entity.getMessage(), type,
                                    TakeOutOrderType.EleMe.getValue());
                        } else if (entity.getType() == 56) {
                            dealOrder(TakeOutOrderStatus.SETTLED.getValue(), entity.getMessage(), type,
                                    TakeOutOrderType.EleMe.getValue());
                        }
                    } else if (StringUtils.equalsIgnoreCase(PushMsgType.POI_STATUS.getValue(), message.getTag())) {
                        dealPoiStatus(entity);
                    } else if (type == PushMsgType.ORDER_REVOKE_REFUND) {
                        dealOrder(TakeOutOrderStatus.REVOKE_REFUND.getValue(), entity.getMessage(), type,
                                TakeOutOrderType.EleMe.getValue());
                    }

                } catch (Exception e) {
                    log.error("EleOrderConsumer error", e);
                    return Action.ReconsumeLater;
                }
                return Action.CommitMessage;
            }
        });
        consumer.start();
        log.info("consumer started");
    }

    private void dealPoiStatus(ElePushEntity entity) {
        String msg = entity.getMessage();
        if (StringUtils.isBlank(msg)) {
            return;
        }
        @SuppressWarnings("unchecked")
        Map<String, Object> map = JsonHelper.fromJson(msg, Map.class);

        String shopId = MapUtils.getLong(map, "shopId") + "";
        @SuppressWarnings("unchecked")
        Map<String, Object> newStatus = (Map<String, Object>) MapUtils.getMap(map, "newStatus");
        String level = MapUtils.getString(newStatus, "busy_level");
        // String valid = MapUtils.getString(newStatus, "is_valid");
        POIStatus status = null;
        if ("0".equals(level)) {
            status = POIStatus.ONLINE;
        } else if ("2".equals(level)) {
            status = POIStatus.OFFLINE;
        }
        if (status == null) {
            return;
        }
        @SuppressWarnings("unchecked")
        Map<String, Object> operator = (Map<String, Object>) MapUtils.getMap(map, "operator");

        POIStatusOperator opp = POIStatusOperator.UNKOWN;
        if (operator != null) {
            String op = MapUtils.getString(operator, "type");
            opp = POIStatusOperator.ele(op);
        }
        publishShopStatus(shopId, TakeOutOrderType.EleMe, status, opp, "");
    }

    private void createOrder(String content, ElePushEntity entity) {
        TakeOutOrder order = new TakeOutOrder();
        String msg = entity.getMessage();
        @SuppressWarnings({ "unchecked" })
        Map<String, Object> msgObj = JsonHelper.fromJson(msg, Map.class);
        order.setShopId(entity.getShopId() + "");
        convert(order, msgObj);

        String phone = CollectionUtils.isEmpty(order.getPhoneList()) ? order.getPhone() : order.getPhoneList().get(0);
        User user = null;
        if (StringUtils.isNotBlank(phone)) {
            AuserShop shop = orderMapper.getShop(order.getShopId(), order.getType());
            user = userMapper.getUser(phone, shop.getAuserId());
            if (user == null) {
                user = new User();
                user.setSource(TakeOutOrderType.EleMe.getValue());
                user.setPhone(phone);
                user.setCreateAt(new Date());
                user.setName(order.getConsigness());
                user.setEle_money(BigDecimal.valueOf(order.getTotalPrice()));
                user.setIntegral(BigDecimal.valueOf(order.getTotalPrice()).intValue());
                user.setMt_money(BigDecimal.ZERO);
                user.setAuserId(shop.getAuserId());
                userMapper.insertUser(user);
            }
        }
        order.setUserId(user != null ? user.getId() : 0l);

        TakeOutOrder too = orderMapper.getByPlfOrderIdAndType(order.getPlfOrderId(), order.getType());
        if (too != null) {
            log.info("EleOrderConsumer entity,reqId:{} is duplicate msg", entity.getRequestId());
        } else {

            orderMapper.insertOrder(order);
            List<TakeOutOrderGroup> groups = order.getGroups();
            if (groups != null) {
                for (TakeOutOrderGroup group : groups) {
                    group.setOrderId(order.getId());
                    orderMapper.insertGroup(group);
                    List<TakeOutOrderGroupItem> items = group.getItems();
                    if (items != null) {
                        for (TakeOutOrderGroupItem item : items) {
                            item.setGroupId(group.getId());
                            item.setOrderId(group.getOrderId());
                            orderMapper.insertGroupItem(item);
                        }
                    }
                }
            }

            TakeOutOrderBackup orderBackup = new TakeOutOrderBackup();
            orderBackup.setOrderId(order.getId());
            orderBackup.setContent(content);
            orderMapper.insertOrderBackup(orderBackup);
            publish(order, PushMsgType.ORDER_CREATED, false);
        }
    }

    @SuppressWarnings("unchecked")
    private void convert(TakeOutOrder order, Map<String, Object> msgObj) {
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss");
        order.setType(TakeOutOrderType.EleMe.getValue());
        order.setPlfOrderId((String) msgObj.get("orderId"));
        order.setPlfOrderCode(order.getPlfOrderId());
        order.setPlfCreateAt(getDate(sdf, (String) msgObj.get("createdAt")));
        order.setActiveAt(getDate(sdf, (String) msgObj.get("activeAt")));
        order.setCreateAt(new Date());
        order.setDeliverFee(JsonObjectUtil.getDouble(msgObj.get("deliverFee")));
        // FIXME 待定
        order.setThirdDeliver(true);
        order.setDeliverTime(getDate(sdf, (String) msgObj.get("deliverTime")));
        order.setDescription((String) msgObj.get("description"));
        order.setInvoice((String) msgObj.get("invoice"));
        order.setBook(JsonObjectUtil.getBoolean(msgObj.get("book")));
        order.setPayType(JsonObjectUtil.getBoolean(msgObj.get("onlinePaid")) ? TakeOutOrder.PAY_TYPE_PAID
                : TakeOutOrder.PAY_TYPE_NOTPAID);
        order.setPhoneList((List<String>) msgObj.get("phoneList"));
        order.setPhone(JsonObjectUtil.getListStr(msgObj.get("phoneList")));
        order.setShopId(JsonObjectUtil.getLong(msgObj.get("shopId")) + "");
        order.setShopName((String) msgObj.get("shopName"));
        order.setDaySn(JsonObjectUtil.getInt(msgObj.get("daySn")));
        order.setStatus((String) msgObj.get("status"));
        order.setRefundStatus((String) msgObj.get("refundStatus"));
        order.setPlfUserId((JsonObjectUtil.getLong(msgObj.get("userId"))));
        order.setTotalPrice(JsonObjectUtil.getDouble(msgObj.get("totalPrice")));
        order.setOriPrice(JsonObjectUtil.getDouble(msgObj.get("originalPrice")));
        order.setConsigness((String) msgObj.get("consignee"));
        String geo = (String) msgObj.get("deliveryGeo");

        if (StringUtils.isNotBlank(geo)) {
            String[] arr = geo.split(",");
            if (arr.length >= 2) {
                order.setDeliverLon(NumberUtils.toDouble(arr[0]));
                order.setDeliverLat(NumberUtils.toDouble(arr[1]));
                AuserShop s = orderMapper.getShop(order.getShopId(), order.getType());
                if (s != null) {
                    order.setAuserId(s.getAuserId().longValue());
                    order.setDist(GeoUtil.calDist(order.getDeliverLon(), order.getDeliverLat(), s.getLon(), s.getLat()));
                }
            }
        }
        order.setDeliverPoi((String) msgObj.get("deliveryPoiAddress"));
        order.setInvoiced(JsonObjectUtil.getBoolean(msgObj.get("invoiced")));
        order.setIncome(JsonObjectUtil.getDouble(msgObj.get("income")));
        order.setServiceFee(Math.abs(JsonObjectUtil.getDouble(msgObj.get("serviceFee"))));
        order.setServiceRate(JsonObjectUtil.getDouble(msgObj.get("serviceRate")));
        order.setHongbao(Math.abs(JsonObjectUtil.getDouble(msgObj.get("hongbao"))));
        order.setPackageFee(JsonObjectUtil.getDouble(msgObj.get("packageFee")));
        order.setActivityTotal(Math.abs(JsonObjectUtil.getDouble(msgObj.get("activityTotal"))));
        order.setShopPart(Math.abs(JsonObjectUtil.getDouble(msgObj.get("shopPart"))));
        // 红包费用算到总的平台补贴部分去
        order.setPlfPart(Math.abs(JsonObjectUtil.getDouble(msgObj.get("elemePart"))) + order.getHongbao());
        // build Groups
        List<Map<String, Object>> groupList = (List<Map<String, Object>>) msgObj.get("groups");
        if (CollectionUtils.isEmpty(groupList)) {
            return;
        }
        List<TakeOutOrderGroup> groups = new ArrayList<>();
        order.setGroups(groups);
        for (Map<String, Object> groupMap : groupList) {
            TakeOutOrderGroup group = new TakeOutOrderGroup();
            groups.add(group);
            group.setName((String) groupMap.get("name"));
            group.setType((String) groupMap.get("type"));
            List<Map<String, Object>> itemList = (List<Map<String, Object>>) groupMap.get("items");
            if (CollectionUtils.isEmpty(itemList)) {
                continue;
            }

            List<TakeOutOrderGroupItem> items = new ArrayList<>();
            group.setItems(items);
            for (Map<String, Object> itemMap : itemList) {
                TakeOutOrderGroupItem item = new TakeOutOrderGroupItem();
                items.add(item);
                item.setQuantity(JsonObjectUtil.getInt(itemMap.get("quantity")));
                item.setBarCode((String) itemMap.get("barCode"));
                item.setExtendCode((String) itemMap.get("extendCode"));
                item.setTotal(JsonObjectUtil.getDouble(itemMap.get("total")));
                item.setPrice(JsonObjectUtil.getDouble(itemMap.get("price")));
                item.setSkuId(JsonObjectUtil.getLong(itemMap.get("skuId")) + "");
                item.setName((String) itemMap.get("name"));
                item.setCategoryId(JsonObjectUtil.getInt(itemMap.get("categoryId")) + "");
                item.setAdditions(JsonObjectUtil.getListStr(itemMap.get("additions")));
                try {
                    List<Map<String, String>> attrs = (List<Map<String, String>>) itemMap.get("attributes");
                    if (!CollectionUtils.isEmpty(attrs)) {
                        List<String> attrList = new ArrayList<>();
                        for (Map<String, String> m : attrs) {
                            attrList.add(m.get("value"));
                        }
                        item.setAttributes(StringUtils.join(attrList, "+"));
                    }
                    List<Map<String, String>> newSpecs = (List<Map<String, String>>) itemMap.get("newSpecs");
                    if (!CollectionUtils.isEmpty(newSpecs)) {
                        List<String> specList = new ArrayList<>();
                        for (Map<String, String> m : newSpecs) {
                            specList.add(m.get("value"));
                        }
                        item.setNewSpecs(StringUtils.join(specList, "+"));
                    }
                } catch (Exception e) {
                    log.error("attr error", e);
                }
                // item.setNewSpecs(JsonObjectUtil.getListStr(itemMap.get("newSpecs")));
                item.setWeight(JsonObjectUtil.getDouble(itemMap.get("weight")));
            }
        }
    }

    public static void main(String[] args) {
        // String msg =
        // "{\"requestId\":\"100006153621019829\",\"type\":10,\"appId\":59816282,\"message\":\"{\\\"id\\\":\\\"1211735509702688813\\\",\\\"orderId\\\":\\\"1211735509702688813\\\",\\\"address\\\":\\\"海兴花园马壮街26号
        // 2号楼4单元13楼2门\\\",\\\"createdAt\\\":\\\"2017-09-07T21:50:17\\\",\\\"activeAt\\\":\\\"2017-09-07T21:50:17\\\",\\\"deliverFee\\\":6.0,\\\"deliverTime\\\":null,\\\"description\\\":\\\"2用餐\\\",\\\"groups\\\":[{\\\"name\\\":\\\"1号篮子\\\",\\\"type\\\":\\\"normal\\\",\\\"items\\\":[{\\\"id\\\":697973075,\\\"skuId\\\":315084428469,\\\"name\\\":\\\"台湾无骨鸡柳\\\",\\\"categoryId\\\":1,\\\"price\\\":6.0,\\\"quantity\\\":1,\\\"total\\\":6.0,\\\"additions\\\":[],\\\"newSpecs\\\":[],\\\"attributes\\\":[],\\\"extendCode\\\":\\\"\\\",\\\"barCode\\\":\\\"\\\",\\\"weight\\\":1.0,\\\"userPrice\\\":0.0,\\\"shopPrice\\\":0.0},{\\\"id\\\":697972604,\\\"skuId\\\":315083946165,\\\"name\\\":\\\"肥牛土豆粉[不辣]\\\",\\\"categoryId\\\":1,\\\"price\\\":18.0,\\\"quantity\\\":1,\\\"total\\\":18.0,\\\"additions\\\":[],\\\"newSpecs\\\":[{\\\"name\\\":\\\"规格\\\",\\\"value\\\":\\\"肥牛土豆粉\\\"}],\\\"attributes\\\":[{\\\"name\\\":\\\"辣度\\\",\\\"value\\\":\\\"不辣\\\"}],\\\"extendCode\\\":\\\"\\\",\\\"barCode\\\":\\\"\\\",\\\"weight\\\":1.0,\\\"userPrice\\\":0.0,\\\"shopPrice\\\":0.0},{\\\"id\\\":697985053,\\\"skuId\\\":315096693941,\\\"name\\\":\\\"桂花肠土豆粉[不辣]\\\",\\\"categoryId\\\":1,\\\"price\\\":14.0,\\\"quantity\\\":1,\\\"total\\\":14.0,\\\"additions\\\":[],\\\"newSpecs\\\":[{\\\"name\\\":\\\"规格\\\",\\\"value\\\":\\\"桂花肠土豆粉\\\"}],\\\"attributes\\\":[{\\\"name\\\":\\\"辣度\\\",\\\"value\\\":\\\"不辣\\\"}],\\\"extendCode\\\":\\\"\\\",\\\"barCode\\\":\\\"\\\",\\\"weight\\\":1.0,\\\"userPrice\\\":0.0,\\\"shopPrice\\\":0.0}]},{\\\"name\\\":\\\"其它费用\\\",\\\"type\\\":\\\"extra\\\",\\\"items\\\":[{\\\"id\\\":-70000,\\\"skuId\\\":-1,\\\"name\\\":\\\"餐盒\\\",\\\"categoryId\\\":102,\\\"price\\\":2.0,\\\"quantity\\\":1,\\\"total\\\":2.0,\\\"additions\\\":[],\\\"newSpecs\\\":null,\\\"attributes\\\":null,\\\"extendCode\\\":\\\"\\\",\\\"barCode\\\":\\\"\\\",\\\"weight\\\":null,\\\"userPrice\\\":0.0,\\\"shopPrice\\\":0.0}]},{\\\"name\\\":\\\"赠品\\\",\\\"type\\\":\\\"discount\\\",\\\"items\\\":[{\\\"id\\\":74933993,\\\"skuId\\\":-1,\\\"name\\\":\\\"大连汽水1份\\\",\\\"categoryId\\\":12,\\\"price\\\":0.0,\\\"quantity\\\":1,\\\"total\\\":0.0,\\\"additions\\\":[],\\\"newSpecs\\\":null,\\\"attributes\\\":null,\\\"extendCode\\\":\\\"\\\",\\\"barCode\\\":\\\"\\\",\\\"weight\\\":null,\\\"userPrice\\\":0.0,\\\"shopPrice\\\":0.0}]}],\\\"invoice\\\":null,\\\"book\\\":false,\\\"onlinePaid\\\":true,\\\"railwayAddress\\\":null,\\\"phoneList\\\":[\\\"13841302245\\\"],\\\"shopId\\\":156871225,\\\"shopName\\\":\\\"金秋园土豆粉\\\",\\\"daySn\\\":2,\\\"status\\\":\\\"unprocessed\\\",\\\"refundStatus\\\":\\\"noRefund\\\",\\\"userId\\\":13954526,\\\"totalPrice\\\":28.0,\\\"originalPrice\\\":46.0,\\\"consignee\\\":\\\"李(女士)\\\",\\\"deliveryGeo\\\":\\\"123.34491989,41.79803998\\\",\\\"deliveryPoiAddress\\\":\\\"海兴花园马壮街26号
        // 2号楼4单元13楼2门\\\",\\\"invoiced\\\":false,\\\"income\\\":18.04,\\\"serviceRate\\\":0.18,\\\"serviceFee\\\":-3.96,\\\"hongbao\\\":0.0,\\\"packageFee\\\":2.0,\\\"activityTotal\\\":-18.0,\\\"shopPart\\\":-18.0,\\\"elemePart\\\":-0.0,\\\"downgraded\\\":false,\\\"vipDeliveryFeeDiscount\\\":0.0,\\\"openId\\\":\\\"\\\",\\\"secretPhoneExpireTime\\\":null,\\\"orderActivities\\\":[{\\\"categoryId\\\":12,\\\"name\\\":\\\"在线支付立减优惠\\\",\\\"amount\\\":-18.0,\\\"elemePart\\\":0.0,\\\"restaurantPart\\\":0.0,\\\"id\\\":82161465},{\\\"categoryId\\\":12,\\\"name\\\":\\\"大连汽水1份\\\",\\\"amount\\\":0.0,\\\"elemePart\\\":0.0,\\\"restaurantPart\\\":0.0,\\\"id\\\":74933993}],\\\"invoiceType\\\":null,\\\"taxpayerId\\\":\\\"\\\"}\",\"shopId\":156871225,\"timestamp\":1504792217743,\"signature\":\"F9B09BA84376B5AC993187426C52746E\",\"userId\":256908975230490863}";
        // TakeOutOrder order = new TakeOutOrder();
        // @SuppressWarnings({ "unchecked" })
        // Map<String, Object> msgObj = JsonHelper.fromJson(msg, Map.class);
        // convert(order, JsonHelper.fromJson((String) msgObj.get("message"),
        // Map.class));
        // System.out.println(JsonHelper.toJson(order));

        // Properties properties = new Properties();
        // properties.put(PropertyKeyConst.ProducerId, "PID_super_test");
        // properties.put(PropertyKeyConst.AccessKey, "LTAIayggS1c9jA2q");
        // properties.put(PropertyKeyConst.SecretKey,
        // "u7XDsuMBjLvkzdn3o6Sct7PhEQQ5Kw");
        //
        // Producer eleProducer = ONSFactory.createProducer(properties);
        // eleProducer.start();
        // for (int i = 1000; i < 2000; i++) {
        // System.out.println(eleProducer.send(new Message("super_test", "aaa",
        // ("" + i).getBytes())).getMessageId());
        // }
        // // Thread.sleep(5000);
        // // System.out.println(eleProducer.send(new Message("ele_order",
        // "bbb",
        // // "bbb".getBytes())).getMessageId());
        // // // Thread.sleep(5000);
        // // System.out.println(eleProducer.send(new Message("ele_order",
        // "aaa",
        // // "aaa".getBytes())).getMessageId());
        // // // Thread.sleep(5000);
        // // System.out.println(eleProducer.send(new Message("ele_order",
        // "bbb",
        // // "bbb".getBytes())).getMessageId());
        // eleProducer.shutdown();
    }

    private static Date getDate(SimpleDateFormat sdf, String time) {
        if (StringUtils.isBlank(time)) {
            return null;
        }
        try {
            return sdf.parse(time);
        } catch (ParseException e) {
        }
        return null;
    }

    private void init() {
        Properties properties = new Properties();
        properties.put(PropertyKeyConst.ConsumerId, consumerId);
        properties.put(PropertyKeyConst.AccessKey, aliyunAccessKey);
        properties.put(PropertyKeyConst.SecretKey, aliyunSecretKey);
        consumer = ONSFactory.createConsumer(properties);
    }
}
